package com.afdor.rws.event.driven.channel.local.producer;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.event.driven.api.channel.producer.EventProducer;
import com.afdor.rws.event.driven.channel.local.observer.ObserverEventSource;
import com.afdor.rws.event.driven.config.EventProducerConfig;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.support.channel.producer.AbstractEventProducer;
import com.afdor.rws.event.driven.support.holder.CallbackEvent;
import com.afdor.rws.event.driven.support.holder.EventResult;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;

/**
 * 本地事件生产者
 *
 * @author 悭梵
 * @date Created in 2018-08-08 15:21
 */
@Slf4j
@NoArgsConstructor
@Extension(value = "local", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = EventProducer.class)
public class LocalEventProducer extends AbstractEventProducer<EventProducerConfig> {

    @Getter
    private ObserverEventSource eventSource;

    public LocalEventProducer(ObserverEventSource eventSource) {
        super();
        setEventSource(eventSource);
    }

    @Override
    protected void doDestroy() {
        this.eventSource = null;
    }

    public void setEventSource(ObserverEventSource eventSource) {
        this.eventSource = Optional.ofNullable(eventSource).orElseThrow(() -> new IllegalArgumentException("事件管理容器为空"));
    }

    @Override
    public <V> EventResult<V> sendEvent(Event event) {
        checkup();
        event.setSendType(Event.SendType.SYNC);
        event.setStatus(Event.Status.SEND);
        return getEventSource().sendEvent(event);
    }

    @Override
    public void sendAsyncEvent(Event event) {
        checkup();
        if (Objects.isNull(event.getSendType())) {
            event.setSendType(Event.SendType.ASYNC);
        }
        event.setStatus(Event.Status.SEND);
        getEventSource().sendAsyncEvent(event);
    }

    @Override
    public <V> void asyncSendEvent(@NonNull CallbackEvent<V> callbackEvent) {
        checkup();
        try {
            Event event = callbackEvent.getEvent();
            event.setSendType(Event.SendType.ASYNC_CALLBACK);
            event.setStatus(Event.Status.SEND);
            CompletableFuture<EventResult<V>> completableFuture = getEventSource().asyncSendEvent(callbackEvent.getEvent());
            callbackEvent.complete(completableFuture.get());
        } catch (Exception e) {
            callbackEvent.failureExceptionally(e);
        }
    }

    @Override
    public void batchSendAsyncEvent(List<Event> eventList) {
        ObserverEventSource eventSource = getEventSource();
        eventList.forEach(event -> {
            if (Objects.isNull(event.getSendType())) {
                event.setSendType(Event.SendType.ASYNC_BATCH);
            }
            event.setStatus(Event.Status.SEND);
            eventSource.sendAsyncEvent(event);
        });
    }
}
